package com.atguigu.day04;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.UrlCountPerWindow;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

//统计的是每个url在 每个5秒钟的滚动窗口中 的被访问次数
//全窗口聚合函数
public class Example5 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                //5秒钟的滚动窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new WindowResult())
                .print();


        env.execute();
    }

    public static class WindowResult extends ProcessWindowFunction<ClickEvent, UrlCountPerWindow,String, TimeWindow>{
        //迭代器elements中包含了属于窗口的所有元素
        //窗口闭合时触发process函数的调用
        @Override
        public void process(String key, Context context, Iterable<ClickEvent> elements, Collector<UrlCountPerWindow> out) throws Exception {

            String url = key;
            Long count = 0L;
            for (ClickEvent e : elements) count++;
            Long windowStart = context.window().getStart();
            Long windowEnd = context.window().getEnd();
            out.collect(new UrlCountPerWindow(url,count,windowStart,windowEnd));
        }
    }





}
